package org.budo.dubbo.registry.redis;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;

import org.budo.dubbo.registry.redis.util.BudoRedisRegistryRepository;
import org.budo.dubbo.registry.redis.util.RegistryJedisUtil;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.ConcurrentHashSet;
import com.alibaba.dubbo.common.utils.NamedThreadFactory;
import com.alibaba.dubbo.registry.NotifyListener;
import com.alibaba.dubbo.registry.support.FailbackRegistry;
import com.alibaba.dubbo.registry.support.SkipFailbackWrapperException;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

/**
 * @author li
 * @version 1 (2015年11月14日 上午11:10:55)
 * @since Java7
 */
public abstract class AbstractBudoRedisRegistry extends FailbackRegistry {
    private static final Logger log = LoggerFactory.getLogger(AbstractBudoRedisRegistry.class);

    private static final ExecutorService JEDIS_SUBSCRIBE_EXECUTOR = Executors.newFixedThreadPool(1);

    public static final String PUBLISH_CHANNEL_PREFIX = "BudoRedisRegistry.";

    private static JedisPool staticJedisPool;

    private static BudoRedisRegistryPubSub staticJedisRegistryPubSub;

    final ScheduledExecutorService expireExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("DubboRegistryExpireTimer", true));

    ScheduledFuture<?> expireFuture;

    BudoRedisRegistryRepository redisRegistryRepository = new BudoRedisRegistryRepository();

    public AbstractBudoRedisRegistry(URL url) {
        super(url);
    }

    /**
     * @see com.alibaba.dubbo.registry.support.FailbackRegistry#register(URL)
     */
    void failbackRegistry_register(URL url) {
        this.abstractRegistry_register(url);

        this.getFailedRegistered().remove(url);
        this.getFailedUnregistered().remove(url);

        try {
            // 向服务器端发送注册请求
            this.doRegister(url);
        } catch (Exception e) {
            Throwable t = e;

            // 如果开启了启动时检测，则直接抛出异常
            boolean check = this.getUrl().getParameter(Constants.CHECK_KEY, true) //
                    && url.getParameter(Constants.CHECK_KEY, true) //
                    && !Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());

            boolean skipFailback = t instanceof SkipFailbackWrapperException;
            if (check || skipFailback) {
                if (skipFailback) {
                    t = t.getCause();
                }
                throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
            }
            log.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);

            // 将失败的注册请求记录到失败列表，定时重试
            this.getFailedRegistered().add(url);
        }
    }

    /**
     * @see com.alibaba.dubbo.registry.support.AbstractRegistry#register(URL)
     */
    private void abstractRegistry_register(URL url) {
        if (url == null) {
            throw new IllegalArgumentException("register url == null");
        }

        if (log.isDebugEnabled()) { // 原来是info
            log.debug("Register: " + url);
        }

        this.getRegistered().add(url);
    }

    /**
     * @see com.alibaba.dubbo.registry.support.FailbackRegistry#subscribe(URL, NotifyListener)
     */
    void failbackRegistry_subscribe(URL url, NotifyListener listener) {
        this.abstractRegistry_subscribe(url, listener); // 原为 super.subscribe(url, listener);

        this.failbackRegistry_removeFailedSubscribed(url, listener); // FailbackRegistry.removeFailedSubscribed(URL, NotifyListener)

        try { // 向服务器端发送订阅请求
            this.doSubscribe(url, listener);
        } catch (Exception e) {
            Throwable t = e;

            List<URL> urls = this.getCacheUrls(url);
            if (urls != null && urls.size() > 0) {
                this.notify(url, listener, urls);
                log.error("Failed to subscribe " + url + ", Using cached list: " + urls + " from cache file: " + this.getUrl().getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/dubbo-registry-" + url.getHost() + ".cache") + ", cause: " + t.getMessage(), t);
            } else { // 如果开启了启动时检测，则直接抛出异常
                boolean check = this.getUrl().getParameter(Constants.CHECK_KEY, true) && url.getParameter(Constants.CHECK_KEY, true);

                boolean skipFailback = t instanceof SkipFailbackWrapperException;
                if (check || skipFailback) {
                    if (skipFailback) {
                        t = t.getCause();
                    }
                    throw new IllegalStateException("Failed to subscribe " + url + ", cause: " + t.getMessage(), t);
                }
                log.error("Failed to subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
            }

            // 将失败的订阅请求记录到失败列表，定时重试
            this.failbackRegistry_addFailedSubscribed(url, listener);
        }
    }

    /**
     * @see com.alibaba.dubbo.registry.support.AbstractRegistry#subscribe(URL, NotifyListener)
     */
    void abstractRegistry_subscribe(URL url, NotifyListener listener) {
        if (url == null) {
            throw new IllegalArgumentException("subscribe url == null");
        }
        if (listener == null) {
            throw new IllegalArgumentException("subscribe listener == null");
        }

        if (log.isDebugEnabled()) { // 原来是info 级别
            log.debug("Subscribe: " + url);
        }

        Set<NotifyListener> listeners = this.getSubscribed().get(url);
        if (listeners == null) {
            ((ConcurrentMap<URL, Set<NotifyListener>>) this.getSubscribed()).putIfAbsent(url, new ConcurrentHashSet<NotifyListener>());
            listeners = this.getSubscribed().get(url);
        }

        listeners.add(listener);
    }

    /**
     * @see com.alibaba.dubbo.registry.support.FailbackRegistry#addFailedSubscribed(URL, NotifyListener)
     */
    void failbackRegistry_addFailedSubscribed(URL url, NotifyListener listener) {
        Set<NotifyListener> listeners = this.getFailedSubscribed().get(url);
        if (listeners == null) {
            ((ConcurrentMap<URL, Set<NotifyListener>>) this.getFailedSubscribed()).putIfAbsent(url, new ConcurrentHashSet<NotifyListener>());
            listeners = this.getFailedSubscribed().get(url);
        }
        listeners.add(listener);
    }

    /**
     * @see com.alibaba.dubbo.registry.support.FailbackRegistry#removeFailedSubscribed(URL, NotifyListener)
     */
    void failbackRegistry_removeFailedSubscribed(URL url, NotifyListener listener) {
        Set<NotifyListener> listeners = this.getFailedSubscribed().get(url);
        if (listeners != null) {
            listeners.remove(listener);
        }

        listeners = this.getFailedUnsubscribed().get(url);
        if (listeners != null) {
            listeners.remove(listener);
        }

        Map<NotifyListener, List<URL>> notified = this.getFailedNotified().get(url);
        if (notified != null) {
            notified.remove(listener);
        }
    }

    BudoRedisRegistryPubSub getJedisRegistryPubSub() {
        if (null == staticJedisRegistryPubSub) {
            this.initJedisRegistryPubSub();
        }

        return staticJedisRegistryPubSub;
    }

    private synchronized void initJedisRegistryPubSub() {
        if (null == staticJedisRegistryPubSub) {
            staticJedisRegistryPubSub = new BudoRedisRegistryPubSub(this);

            JEDIS_SUBSCRIBE_EXECUTOR.submit(new Runnable() {
                public void run() {
                    while (true) {
                        AbstractBudoRedisRegistry.this.jedisSubscribe(); // 应该会阻塞
                    }
                }
            });
        }
    }

    /**
     * 正常情况下应当会阻塞, 方法不会返回
     */
    private void jedisSubscribe() {
        Jedis listenerJedis = null;
        try {
            listenerJedis = RegistryJedisUtil.getJedisResource(this);
            log.info("#223 listenerJedisPsubscribe, listenerJedis=" + listenerJedis);

            listenerJedis.psubscribe(staticJedisRegistryPubSub, PUBLISH_CHANNEL_PREFIX + "*"); // 订阅redis 这里会阻塞, 不能归还jedis连接
        } catch (Throwable e) {
            if (("" + e).contains("JedisConnectionException")) {
                log.error("#228 JedisConnectionException error, sleep 5 seconds, e=" + e);
                this.threadSleep(5 * 1000);
            } else {
                log.error("#230 listenerJedisPsubscribe error, e=" + e, e);
            }
        } finally {
            log.info("#233 return listenerJedis, listenerJedis=" + listenerJedis);

            RegistryJedisUtil.returnResource(this, listenerJedis);
        }
    }

    private void threadSleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (Throwable e) {
            log.error("#245 threadSleep error, e=" + e, e);
        }
    }

    public JedisPool getJedisPool() {
        if (null == staticJedisPool) {
            synchronized (AbstractBudoRedisRegistry.class) {
                if (null == staticJedisPool) {
                    staticJedisPool = RegistryJedisUtil.initJedisPool(this);
                }
            }
        }

        return staticJedisPool;
    }

    void cancelExpireFuture(boolean mayInterruptIfRunning) {
        try {
            expireFuture.cancel(mayInterruptIfRunning);
        } catch (Throwable e) {
            log.error("cancelExpireFuture error, e=" + e, e);
        }
    }

    public String getProtocol() {
        if (null == this.getUrl()) {
            return null;
        }

        return this.getUrl().getProtocol();
    }
}